package org.systemsbiology.rface.hadoop.mappers;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.systemsbiology.hadoop.ioformats.LexicalKey;
import org.systemsbiology.rface.hadoop.Main;
import org.systemsbiology.rface.hadoop.data.ResultGroup;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.regex.Pattern;

/**
 * Created by IntelliJ IDEA.
 * User: anorberg
 * Date: 9/19/11
 * Time: 11:54 AM
 * To change this template use File | Settings | File Templates.
 */
public class ResultFromFileLine extends Mapper<Text, Text, LexicalKey, ResultGroup> {

    private int[] orderedIndexes;
    private int[] unorderedIndexes;
    private int[] targetIndexes;
    private HashSet<String> interestingFiles;
    private static final HashSet<String> INFINITIES;
    static{
        INFINITIES=new HashSet<String>();
        INFINITIES.add("inf");
        INFINITIES.add("+inf");
        INFINITIES.add("infinity");
        INFINITIES.add("0.infinity");
        INFINITIES.add("+0.infinity");
        INFINITIES.add("1.#inf");
        INFINITIES.add("+1.#inf");
    }

    public ResultFromFileLine(){}

    private static int[] stringsToInts(String[] strings){
        int[] ret = new int[strings.length];
        for(int k = 0; k < strings.length; ++k){
            ret[k] = Integer.parseInt(strings[k]);
        }
        return ret;
    }

    @Override
    protected void setup(Context context){
        //calculate column offsets from configuration records
        Configuration config = context.getConfiguration();
        String[] orderedDexStr = config.get("OrderedKeyIndexes").split("\\s");
        String[] unorderedDexStr = config.get("UnorderedKeyIndexes").split("\\s");
        String[] targetDexStr=config.get("Targets").split("\\s");

        orderedIndexes = stringsToInts(orderedDexStr);
        unorderedIndexes = stringsToInts(unorderedDexStr);
        targetIndexes = stringsToInts(targetDexStr);

        interestingFiles = new HashSet<String>();
        for(String filename:config.get(Main.INTERESTING_FILES_KEY).split(Pattern.quote(File.pathSeparator))){
            interestingFiles.add(filename);
        }
    }

    @Override
    protected void map(Text key, Text value, Context context) throws IOException, InterruptedException{

        //path-aware interesting-path-or-file name matching algorithm
        String[] progressiveNames = key.toString().split(Pattern.quote(File.separator));
        StringBuilder root = new StringBuilder();
        for(int k = 0; k < key.getLength(); ++k){
            if(key.charAt(k) == File.separatorChar){
                root.append(File.separator);
            } else {
                break;
            }
        }
        boolean first = true;
        boolean isInteresting = false;
        for(String s: progressiveNames){
            if(!first){
                root.append(File.separator);
            }
            first = false;
            root.append(s);
            if(interestingFiles.contains(root.toString())){
                isInteresting=true;
                break;
            }
        }
        String[] record = value.toString().split("\t");
        String[] compoundKey = new String[orderedIndexes.length + unorderedIndexes.length + 1];
        for(int k = 0; k < orderedIndexes.length; ++k){
            compoundKey[k] = record[orderedIndexes[k]];
        }
        String[] unordereds = new String[unorderedIndexes.length];
        for(int k = 0; k < unorderedIndexes.length; ++k){
            unordereds[k] = record[unorderedIndexes[k]];
        }
        Arrays.sort(unordereds);
        for(int k = 0; k < unordereds.length; ++k){
            compoundKey[k + orderedIndexes.length] = unordereds[k];
        }

        for(int k = 0; k < targetIndexes.length; ++k){
            compoundKey[compoundKey.length - 1] = String.valueOf(targetIndexes[k]); //the key ID itself, not its data!
            double targValue = Double.NaN;
            try{
                targValue = Double.parseDouble(record[targetIndexes[k]]);
            } catch(NumberFormatException wrongNumber){
                String s = record[targetIndexes[k]];
                s = s.trim();
                if(s.length()  == 0){
                    targValue = 0;
                    System.err.println("warning: blank value encountered; interpreted as zero");
                } else {
                    boolean negative = false;
                    if(s.charAt(0) == '-'){
                        negative = true;
                        s = s.substring(1);
                    }
                    if(INFINITIES.contains(s.toLowerCase())){
                        targValue = negative ? Double.NEGATIVE_INFINITY : Double.POSITIVE_INFINITY;
                    } else {
                        targValue = Double.NaN;
                        System.err.println("error: unrecognizable number "+record[targetIndexes[k]]+
                                "; interpreted as not a number");
                    }
                }
            }
            context.write(new LexicalKey(compoundKey), new ResultGroup(targValue, isInteresting));
        }
    }
}
